package eg;

// WholeFileRecordReader The RecordReader used by WholeFileInputFormat for reading a whole file as a record
import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;

// MyCustomRecordReader
class CustomReader extends RecordReader<LongWritable, Text> {
  
  private FileSplit fileSplit;
  private Configuration conf;
  private LongWritable key = new LongWritable();
  private long start;
  private long end;
  private Text value = new Text();
  private boolean processed = false;
  FSDataInputStream in = null;

  @Override
  public void initialize(InputSplit split, TaskAttemptContext context)
      throws IOException, InterruptedException {
    this.fileSplit = (FileSplit) split;
    this.conf = context.getConfiguration();
    
    start = fileSplit.getStart();
    end = start + fileSplit.getLength();
    
    Path file = fileSplit.getPath();
    FileSystem fs = file.getFileSystem(conf);
    in = fs.open(file);
    
    if(start!=0){
        in.seek(start);
    }
  }
  
  @Override
  public boolean nextKeyValue() throws IOException, InterruptedException {
          
        key.set(start);
    if (!processed) {
      byte[] contents = new byte[(int) fileSplit.getLength()+99+1]; 
      try {
        if((int)end==1073741824){//last split 1073741824 for 1GB,1048576 for 1MB
            contents[0]=0;    //add 1 byte to tell add99byte or not
        	IOUtils.readFully(in, contents, 1, contents.length-99-1);
        }else{
        	contents[0]=99;   //add 1 byte to tell add99byte or not
        	IOUtils.readFully(in, contents, 1, contents.length-1);
        }
        value.set(contents, 0, contents.length);
      } finally {
        IOUtils.closeStream(in);
      }
      processed = true;
      return true;
    }
    return false;
  }
  
  @Override
  public LongWritable getCurrentKey() throws IOException, InterruptedException {
    return key;
  }

  @Override
  public Text getCurrentValue() throws IOException,
      InterruptedException {
    return value;
  }

  @Override
  public float getProgress() throws IOException {
    return processed ? 1.0f : 0.0f;
  }

  @Override
  public void close() throws IOException {
    // do nothing
  }
}
//^^ WholeFileRecordReader